package com.bestcem.xm.txtanls.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Validator;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bestcem.xm.bi.grpc.v1.services.Filter;
import com.bestcem.xm.bi.grpc.v1.services.WorktableTtype;
import com.bestcem.xm.common.core.constant.Constants;
import com.bestcem.xm.common.core.domain.web.ServiceResult;
import com.bestcem.xm.txtanls.dao.AnalysisProjectDao;
import com.bestcem.xm.txtanls.dao.ViewpointDao;
import com.bestcem.xm.txtanls.entity.mongo.JoinInfo;
import com.bestcem.xm.txtanls.entity.pojo.ProjectDO;
import com.bestcem.xm.txtanls.entity.pojo.ProjectFieldDO;
import com.bestcem.xm.txtanls.entity.pojo.RecordDO;
import com.bestcem.xm.txtanls.entity.pojo.ViewpointDO;
import com.bestcem.xm.txtanls.enums.AnalysisStatusEnum;
import com.bestcem.xm.txtanls.enums.DatasourceEnum;
import com.bestcem.xm.txtanls.enums.TrainStatusEnum;
import com.bestcem.xm.txtanls.grpc.client.BaseBiGrpcService;
import com.bestcem.xm.txtanls.grpc.client.ClickhouseGrpcService;
import com.bestcem.xm.txtanls.retry.CallAnalysisApiSpringRetry;
import com.bestcem.xm.txtanls.service.AsyncTaskService;
import com.bestcem.xm.txtanls.service.dto.train.ClickHouseDTO;
import com.bestcem.xm.txtanls.service.dto.train.ClickhouseColumnDTO;
import com.bestcem.xm.txtanls.service.dto.train.SentimentDTO;
import com.bestcem.xm.txtanls.service.dto.train.TrainTextDTO;
import com.bestcem.xm.txtanls.service.dto.worktable.ColumnDTO;
import com.bestcem.xm.txtanls.service.dto.worktable.DatasourceDTO;
import com.bestcem.xm.txtanls.service.dto.worktable.WorktableDTO;
import com.bestcem.xm.txtanls.util.AnalysisUtil;
import com.bestcem.xm.txtanls.util.AsyncTaskUtil;
import com.bestcem.xm.txtanls.util.ClickhouseUtil;
import com.bestcem.xm.txtanls.util.TextUtil;
import com.zy.dp.proto.strontium.v1.TextSentiment;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

import static com.bestcem.xm.common.core.constant.Constants.*;

/**
 * 异步任务处理类
 *
 * @author gy.liu
 * @version 1.0
 * @date 2021/5/10 13:34
 **/
@Slf4j
@Service
public class AsyncTaskServiceImpl implements AsyncTaskService {
    public static final String COLLECTION_ID = "collectionId";
    public static final String COLLECTION_TIME = "collectionTime";
    public static final String TIME = "time";
    public static final String SEQ = "seq";
    public static final String ROW_INDEX = "name";
    @Resource
    private ViewpointDao viewPointDao;
    @Resource
    private AnalysisProjectDao analysisProjectDao;
    @Resource
    private CallAnalysisApiSpringRetry callAnalysisApiSpringRetry;
    @Resource
    private AnalysisUtil analysisUtil;
    @Resource
    private AsyncTaskUtil asyncTaskUtil;
    @Resource
    private ClickhouseUtil clickhouseUtil;
    @Resource
    private BaseBiGrpcService baseBiGrpcService;
    @Resource
    private ClickhouseGrpcService clickhouseGrpcService;

    @Override
    public ServiceResult<Object> trainTask(ProjectDO project) {
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] 异步训练开始 {}", project.getId());
        }

        // 训练开始时的那一秒
        String beginTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern(TIME_FORMAT));

        String projectId = project.getId();
        String tableId = project.getTableId();
        if (ObjectUtils.isEmpty(tableId)) {
            log.error("[Txtanls] {} tableId为空", project);
            // tableId 为空时, 不跑训练
            int noneUpdateResult = analysisUtil.updateProjectStatus(project.getId(), AnalysisStatusEnum.FINISHED);
            if (noneUpdateResult == 0) {
                log.error("[Txtanls] {} 空tableId更新训练状态失败", projectId);
            }
            return ServiceResult.success();
        }

        // 获取真实的 worktableId
        String worktableId = tableId;
        if (!ObjectUtils.isEmpty(project.getSurveyId())) {
            worktableId = clickhouseGrpcService.getSurveyWorkTableId(project.getSurveyId());
            if (ObjectUtils.isEmpty(worktableId)) {
                analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.INACTIVE);
                return ServiceResult.success("问卷数据源id查询失败");
            }
        }

        WorktableDTO worktableDTO = baseBiGrpcService.getWorktable(worktableId);
        if (Objects.isNull(worktableDTO)) {
            // 是否为示例项目
            if ("文本分析项目示例".equals(project.getName()) && "餐饮数据源demo".equals(project.getTableName())) {
                // 原有示例项目是 INACTIVE , 改成 FINISHED
                analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.FINISHED);
                return ServiceResult.success();
            }
            log.error("[Txtanls] {} Worktable信息 {} 查询失败", project, worktableId);
            return ServiceResult.fail("Worktable信息查询失败");
        }
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] {} Worktable信息 {} ", worktableId, worktableDTO);
        }
        DatasourceDTO datasourceDTO = baseBiGrpcService.getDataSource(worktableDTO.getDatasourceId());
        if (Objects.isNull(datasourceDTO)) {
            log.error("[Txtanls] {} Datasource信息 {} 查询失败", project, worktableId);
            return ServiceResult.fail("datasource信息查询失败");
        }
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] {} Datasource信息 {} ", worktableId, datasourceDTO);
        }
        // 数据源类型
        int datasourceType = datasourceDTO.getType();

        // 只有 CREATED/INACTIVE 的项目参与训练
        project = analysisProjectDao.findById(projectId);
        if (AnalysisStatusEnum.CREATED.getIndex() != project.getStatus()
                && AnalysisStatusEnum.INACTIVE.getIndex() != project.getStatus()) {
            if (log.isInfoEnabled()) {
                log.info("[Txtanls] {} 项目已在训练 {}", project, Thread.currentThread().getName());
            }
            return ServiceResult.success();
        }

        // 获取全部的列信息
        List<ColumnDTO> columnDTOs = asyncTaskUtil.getColumnInfo(worktableDTO.getId());
        if (log.isInfoEnabled()) {
            log.info("[txtanls]: 文本分析查询问卷数据结果为：{},projectId:{}", JSONObject.toJSONString(columnDTOs), project.getId());
        }
        // 获取全部的列信息和对应匹配字段的对应信息
        Map<String, ColumnDTO> columnDTOMap = asyncTaskUtil.getColumnDTOMap(project, columnDTOs, datasourceType);
        // 获取数据总条数查询clickhouse需要的列信息
        List<ClickhouseColumnDTO> worktableDataTotalColumnList = clickhouseUtil.getWorktableDataTotalColumns(project, columnDTOMap, datasourceType);
        // 获取数据总条数
        long totalData = clickhouseGrpcService.getWorktableDataTotal(worktableId, worktableDataTotalColumnList);
        if (totalData < 0) {
            return ServiceResult.fail("clickhouse 总数据条数查询失败");
        }
        if (totalData == 0) {
            if (log.isInfoEnabled()) {
                log.info("[Txtanls] 实时训练 {} clickhouse 无数据", projectId);
            }
        }

        // 数据表长度 * 要训练的字段数量 = 待训练的条数
        if (!CollectionUtils.isEmpty(project.getTextFieldList())) {
            totalData *= project.getTextFieldList().size();
        } else {
            totalData = 0L;
        }

        if (totalData == 0) {
            if (log.isInfoEnabled()) {
                log.info("[Txtanls] 实时训练 {} clickhouse 无数据", projectId);
            }
        }
        // 已经训练完的条数
        long existedData = asyncTaskUtil.getExistedDataCount(projectId);
        // 待训练的条数
        long remaining = totalData - existedData;
        if (remaining < 0) {
            remaining = 0;
        }

        if (log.isInfoEnabled()) {
            log.info("[Txtanls] 实时训练 {} 总训练条数 {}, 已训练条数 {}, 待训练的条数 {}", projectId, totalData, existedData, remaining);
        }

        // 更新训练状态到初始状态/训练进度为待训练的条数
        int updateStateStartResult = analysisProjectDao.updateAnalysisProjectProgress(projectId, TrainStatusEnum.UNFINISHED, remaining);
        if (updateStateStartResult == 0) {
            log.error("[Txtanls] {} 训练之前更新训练进度失败", projectId);
        }


        // 根据不同的数据源类型做不同的处理
        if (datasourceType == DatasourceEnum.MERGED.getIndex() ||
                datasourceType == DatasourceEnum.CONCAT.getIndex()) { // 合表类型数据源数据处理
            mergeDatasourceHandler(project, worktableDTO, columnDTOs, beginTime);
        } else if (datasourceType == DatasourceEnum.SURVEY.getIndex()) { // 问卷类型数据源数据处理
            surveyDatasourceHandler(project, worktableDTO, columnDTOs, beginTime);
        } else if (datasourceType == DatasourceEnum.UPLOAD.getIndex()) { // 上传类型数据源数据处理
            uploadDatasourceHandler(project, worktableDTO, columnDTOs);
        } else if (datasourceType == DatasourceEnum.TICKET.getIndex() ||
                datasourceType == DatasourceEnum.TICKET_STATUS.getIndex()) { // 工单类型数据源数据处理
            ticketDatasourceHandler(project, worktableDTO, columnDTOs);
        } else { // 其他类型数据源, 当上传类型的来训练
            uploadDatasourceHandler(project, worktableDTO, columnDTOs);
        }

        // 训练完成后更新训练状态
        int updateStateEndResult = analysisProjectDao.updateAnalysisProjectProgress(projectId, TrainStatusEnum.FINISHED, Constants.PROGRESS_FINISHED);
        if (updateStateEndResult == 0) {
            log.error("[Txtanls] {} 训练完成后更新训练状态失败", projectId);
        }
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] 项目 {} 训练完成", projectId);
        }

        return ServiceResult.success();
    }

    /**
     * 合表 数据源训练前置处理
     * 现在大部分的项目都是合表 答卷/工单/问卷
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @param beginTime    本次训练开始时的那一秒
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/9 19:38
     */
    public void mergeDatasourceHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs, String beginTime) {
        String projectId = project.getId();
        // 是否为问卷项目
        if (!ObjectUtils.isEmpty(project.getSurveyId())) {
            surveyTrainHandler(project, worktableDTO, columnDTOs, beginTime);
            // 问卷合表项目训练完成后将状态改为 INACTIVE, 用于后续增量训练标识
            analysisUtil.updateProjectStatus(project.getId(), AnalysisStatusEnum.INACTIVE);
            return;
        }

        // worktable 关联项目id
        String projectIdList = worktableDTO.getProjectIdList();
        JSONArray jsonArray = null;
        if (!ObjectUtils.isEmpty(projectIdList) && projectIdList.length() > 2) {
            try {
                jsonArray = JSONArray.parseArray(projectIdList);
            } catch (Exception e) {
                log.error("[Txtanls] {} 合表项目关联项目id列表解析失败", projectId);
                jsonArray = null;
            }
        }
        // 处理合表中无 答卷/工单 的情况
        if (Objects.isNull(jsonArray) || jsonArray.size() < 1) {
            staticTrainHandler(project, worktableDTO, columnDTOs);
            // 如果合表中都是上传文件，训练完成自动保存为训练完成状态
            analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.FINISHED);
            return;
        }

        // 处理合表中有 答卷/工单 的情况
        incrementTrainHandler(project, worktableDTO, columnDTOs);
        // 问卷项目训练完成后将状态改为 INACTIVE, 用于后续增量训练标识
        analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.INACTIVE);
    }

    /**
     * 问卷/答卷 数据源训练前置处理
     * 可能用不到, 只有在项目没有层级数据源时会用到
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @param beginTime    本次训练开始时的那一秒
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/9 19:38
     */
    public void surveyDatasourceHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs, String beginTime) {
        surveyTrainHandler(project, worktableDTO, columnDTOs, beginTime);
        // 问卷项目训练完成后将状态改为 INACTIVE, 用于后续增量训练标识
        analysisUtil.updateProjectStatus(project.getId(), AnalysisStatusEnum.INACTIVE);
    }

    /**
     * 上传 数据源训练前置处理
     * 可能用不到, 只有在项目没有层级数据源时会用到
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/9 19:38
     */
    public void uploadDatasourceHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs) {
        staticTrainHandler(project, worktableDTO, columnDTOs);
        // 上传项目训练完成后将状态改为 FINISHED
        analysisUtil.updateProjectStatus(project.getId(), AnalysisStatusEnum.FINISHED);
    }

    /**
     * 工单 数据源训练前置处理
     * 可能用不到
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/9 19:38
     */
    public void ticketDatasourceHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs) {
        incrementTrainHandler(project, worktableDTO, columnDTOs);
        // 工单项目训练完成后将状态改为 INACTIVE, 用于后续增量训练标识
        analysisUtil.updateProjectStatus(project.getId(), AnalysisStatusEnum.INACTIVE);
    }

    /**
     * 定量数据 数据源训练
     * 通过记录的最大 collection_id(即为clickhouse查询数据返回的index下标) 来跳过查询
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/23 11:40
     */
    public void staticTrainHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs) {
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] 项目 {} 开始定量数据训练", project.getId());
        }

        String projectId = project.getId();
        String timeFieldName = project.getTimeFieldName();
        // 行业信息
        String business = analysisUtil.getIndustryCode(project.getModelId());
//        IndustryDTO industry = analysisUtil.getIndustryInfo(project.getOrgId());
//
//        if (ObjectUtils.isEmpty(industry)) {
//            analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.FAILED);
//            return;
//        }
        // 获取子观点
        List<ViewpointDO> childrenViewPoint = viewPointDao.findAllChildrenViewPoint(projectId);

        List<ProjectFieldDO> fieldList = project.getTextFieldList();
        if (CollectionUtils.isEmpty(fieldList)) {
            return;
        }
        // 普通项目 field.id == column.id
        Map<String, ColumnDTO> columnDTOMap = columnDTOs.stream()
                .collect(Collectors.toMap(ColumnDTO::getId, self -> self, (prev, now) -> prev));

        for (int textFieldIndex = 0; textFieldIndex < fieldList.size(); textFieldIndex++) {
            ProjectFieldDO field = fieldList.get(textFieldIndex);
            // 项目训练字段的真实字段名
            String fieldName = field.getValue();
            // 项目已有原话记录的最大index
            Integer maxIndex = field.getOffset();
            // 兼容历史方案，如果field中没有offset，就从数据库中查询出最大的Index
            if (Validator.isNull(maxIndex)) {
                maxIndex = asyncTaskUtil.getMaxIndexFromMongo(projectId, field.getId());
            }
            // 待查询的 clickhouse 列信息
            List<ClickhouseColumnDTO> searchColumn = clickhouseUtil.getStaticClickhouseColumns(project, field, columnDTOMap);

            // clickhouse查询的跳跃量 / clickhouse本次查询返回的数据
            int offset = maxIndex, progressCountNew = 0;
            Integer total = Integer.MAX_VALUE;
            List<TrainTextDTO> trainTextDTOList = new ArrayList<>();
            while (offset < total) {
                // 查询clickhouse
                ClickHouseDTO clickHouseDTO = clickhouseUtil.getClickhouseData(worktableDTO.getId(), searchColumn, offset);
                if (Objects.isNull(clickHouseDTO) || CollectionUtils.isEmpty(clickHouseDTO.getAllRowsData())) {
                    log.warn("[Txtanls] {}-{}-{} clickhouse返回结果为空", project.getId(), worktableDTO.getId(), field.getValue());
                    break;
                }
                if (!clickHouseDTO.getColNameList().contains(fieldName)) {
                    log.error("[Txtanls] {}-{}-{} clickhouse返回结果无训练字段", project.getId(), worktableDTO.getId(), field.getValue());
                    break;
                }
                if (!clickHouseDTO.getColNameList().contains(timeFieldName)) {
                    log.error("[Txtanls] {}-{}-{} clickhouse返回结果无时间字段", project.getId(), worktableDTO.getId(), field.getValue());
                    break;
                }
                // clickhouse返回的total
                total = clickHouseDTO.getTotal();
                Integer size = clickHouseDTO.getAllRowsData().size();
                if (log.isInfoEnabled()) {
                    log.info("[Txtanls] 项目id：{}, 查询数据数量：{}，偏移量：{}", project.getId(), total, size);
                }
                for (int index = 0; index < size; index++) {
                    Map<String, String> rowData = clickHouseDTO.getAllRowsData().get(index);
                    // 要训练的文本记录
                    String record = rowData.get(fieldName);
                    // 获取当前数据行的行号[0-999]
                    Integer rowIndex = Integer.parseInt(rowData.getOrDefault(AsyncTaskServiceImpl.ROW_INDEX, "0")) + 1;
                    // 无用字符清理
                    record = TextUtil.text_clean(record);
                    if (!ObjectUtils.isEmpty(record) && !"None".equals(record) && !"null".equals(record)) {
                        // 训练时间字段
                        String time = asyncTaskUtil.formatTime(rowData.get(timeFieldName));
                        trainTextDTOList.add(new TrainTextDTO(record, time, String.valueOf(offset + rowIndex)));
                    }
                    // 到达设置的阈值, 批量训练
                    if (trainTextDTOList.size() == REQUEST_NUMBER) {
                        trainRecordData(project, field, trainTextDTOList, business, childrenViewPoint);
                        trainTextDTOList = new ArrayList<>();
                        // 更新项目中offset的值, 多次更新，减少误差
                        analysisProjectDao.updateProjectFieldOffset(projectId, field.getId(), offset + rowIndex);
                    }
                    // 到达设置的阈值, 更新训练进度
                    if ((progressCountNew += 1) == PROGRESS_COUNT) {
                        asyncTaskUtil.updateTrainProgress(projectId, field.getId(), PROGRESS_COUNT);
                        progressCountNew = 0;
                    }
                }
                // 兜底训练, 防止数据遗漏
                if (!CollectionUtils.isEmpty(trainTextDTOList)) {
                    trainRecordData(project, field, trainTextDTOList, business, childrenViewPoint);
                    // 更新训练进度
                    asyncTaskUtil.updateTrainProgress(projectId, field.getId(), trainTextDTOList.size());
                    trainTextDTOList = new ArrayList<>();
                }

                offset = offset + LIMIT > total ? total : offset + LIMIT;
                // 更新项目中offset的值
                analysisProjectDao.updateProjectFieldOffset(projectId, field.getId(), offset);
            }
        }
    }

    /**
     * 非定量数据 数据源训练
     * 通过记录的最大 collection_id(即为clickhouse查询数据返回的index下标) 来跳过查询
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/23 11:40
     */
    public void incrementTrainHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs) {
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] 项目 {} 开始增量数据训练", project.getId());
        }

        String projectId = project.getId();
        String timeFieldName = project.getTimeFieldName();
        // 行业信息
        String business = analysisUtil.getIndustryCode(project.getModelId());
//        IndustryDTO industry = analysisUtil.getIndustryInfo(project.getOrgId());

        // 获取子观点
        List<ViewpointDO> childrenViewPoint = viewPointDao.findAllChildrenViewPoint(projectId);

        List<ProjectFieldDO> fieldList = project.getTextFieldList();
        if (CollectionUtils.isEmpty(fieldList)) {
            return;
        }
        // 增量项目 field.id == column.id
        Map<String, ColumnDTO> columnDTOMap = columnDTOs.stream()
                .collect(Collectors.toMap(ColumnDTO::getId, self -> self, (prev, now) -> prev));

        // 问卷项目状态过滤
        List<ColumnDTO> collect = columnDTOs.stream().filter(columnDTO -> "status".equals(columnDTO.getCol())).collect(Collectors.toList());
        List<Filter> statusFilter = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(collect) && WorktableTtype.WORKTABLE_TTYPE_SURVEY.getNumber() == worktableDTO.getType()) {
            statusFilter = Collections.singletonList(clickhouseUtil.getSurveyStatusFilter());
        }
        for (int textFieldIndex = 0; textFieldIndex < fieldList.size(); textFieldIndex++) {
            ProjectFieldDO field = fieldList.get(textFieldIndex);
            // 项目训练字段的真实字段名
            String fieldName = field.getValue();
            if (log.isInfoEnabled()) {
                log.info("[Txtanls] 项目字段：{} 开始训练", fieldName);
            }
            // 项目已有原话记录的最大index
            Integer maxIndex = field.getOffset();
            // 兼容历史方案，如果field中没有offset，就从数据库中查询出最大的Index
            if (Validator.isNull(maxIndex)) {
                maxIndex = asyncTaskUtil.getMaxIndexFromMongo(projectId, field.getId());
            }
            // 待查询的 clickhouse 列信息
            List<ClickhouseColumnDTO> searchColumn = clickhouseUtil.getIncrementClickhouseColumns(project, field, columnDTOMap);


            int offset = maxIndex, progressCountNew = 0;
            Integer total = Integer.MAX_VALUE;
            List<TrainTextDTO> trainTextDTOList = new ArrayList<>();
            while (offset < total) {
                // 查询clickhouse
                ClickHouseDTO clickHouseDTO = clickhouseUtil.getClickhouseData(worktableDTO.getId(), searchColumn, offset, statusFilter);
                if (Objects.isNull(clickHouseDTO) || CollectionUtils.isEmpty(clickHouseDTO.getAllRowsData())) {
                    log.warn("[Txtanls] {}-{} clickhouse返回结果为空", worktableDTO.getId(), field.getValue());
                    break;
                }
                if (!clickHouseDTO.getColNameList().contains(fieldName)) {
                    log.error("[Txtanls] {}-{} clickhouse返回结果无训练字段", worktableDTO.getId(), field.getValue());
                    break;
                }
                if (!clickHouseDTO.getColNameList().contains(timeFieldName)) {
                    log.error("[Txtanls] {}-{} clickhouse返回结果无时间字段", worktableDTO.getId(), field.getValue());
                    break;
                }
                total = clickHouseDTO.getTotal();
                int size = clickHouseDTO.getAllRowsData().size();
                if (log.isInfoEnabled()) {
                    log.info("[Txtanls] 项目id：{}, 查询数据数量：{}，偏移量：{}", project.getId(), total, size);
                }
                for (int index = 0; index < clickHouseDTO.getAllRowsData().size(); index++) {
                    Map<String, String> rowData = clickHouseDTO.getAllRowsData().get(index);
                    // 要训练的文本记录
                    String record = rowData.get(fieldName);
                    // 获取当前数据行的行号[0-999]
                    int rowIndex = Integer.parseInt(rowData.getOrDefault(AsyncTaskServiceImpl.ROW_INDEX, "0")) + 1;
                    // 无用字符清理
                    record = TextUtil.text_clean(record);
                    if (!ObjectUtils.isEmpty(record) && !"None".equals(record) && !"null".equals(record)) {
                        // 训练时间字段
                        String time = asyncTaskUtil.formatTime(rowData.get(timeFieldName));
                        trainTextDTOList.add(new TrainTextDTO(record, time, String.valueOf(offset + rowIndex)));
                    }

                    // 到达设置的阈值, 批量训练
                    if (trainTextDTOList.size() == REQUEST_NUMBER) {
                        trainRecordData(project, field, trainTextDTOList, business, childrenViewPoint);
                        trainTextDTOList = new ArrayList<>();
                        // 更新项目中offset的值, 多次更新，减少误差
                        analysisProjectDao.updateProjectFieldOffset(projectId, field.getId(), offset + rowIndex);
                    }
                    // 到达设置的阈值, 更新训练进度
                    if ((progressCountNew += 1) == PROGRESS_COUNT) {
                        asyncTaskUtil.updateTrainProgress(projectId, field.getId(), PROGRESS_COUNT);
                        progressCountNew = 0;
                    }
                }
                // 兜底训练, 防止数据遗漏
                if (!CollectionUtils.isEmpty(trainTextDTOList)) {
                    trainRecordData(project, field, trainTextDTOList, business, childrenViewPoint);
                    // 更新训练进度
                    asyncTaskUtil.updateTrainProgress(projectId, field.getId(), trainTextDTOList.size());
                }
                offset = offset + LIMIT > total ? total : offset + LIMIT;
                // 更新项目中offset的值
                analysisProjectDao.updateProjectFieldOffset(projectId, field.getId(), offset);
            }
        }
    }

    /**
     * 问卷项目 数据源训练
     * 通过记录的最大 collection_time 来跳过查询
     *
     * @param project      项目信息
     * @param worktableDTO worktable信息
     * @param columnDTOs   worktable里所有的列信息
     * @param beginTime    本次训练开始时的那一秒
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/23 11:41
     */
    public void surveyTrainHandler(ProjectDO project, WorktableDTO worktableDTO, List<ColumnDTO> columnDTOs, String beginTime) {
        if (log.isInfoEnabled()) {
            log.info("[Txtanls] 项目 {} 开始问卷数据训练", project.getId());
        }

        String projectId = project.getId();
        String timeFieldName = project.getTimeFieldName();
        // 行业信息
        String business = analysisUtil.getIndustryCode(project.getModelId());
//        IndustryDTO industry = analysisUtil.getIndustryInfo(project.getOrgId());
//        if (ObjectUtils.isEmpty(industry)) {
//            analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.FAILED);
//            return;
//        }
        // 获取子观点
        List<ViewpointDO> childrenViewPoint = viewPointDao.findAllChildrenViewPoint(projectId);

        List<ProjectFieldDO> fieldList = project.getTextFieldList();
        if (CollectionUtils.isEmpty(fieldList)) {
            return;
        }
        // 问卷项目 field.value == column.col
        Map<String, ColumnDTO> columnDTOMap = columnDTOs.stream()
                .collect(Collectors.toMap(ColumnDTO::getCol, self -> self, (prev, now) -> prev));
        // 答卷数据源只取答卷状态为成功完成的数据
        Filter statusFilter = clickhouseUtil.getSurveyStatusFilter();

        for (ProjectFieldDO field : fieldList) {
            // 项目训练字段的真实字段名
            String fieldName = field.getValue();
            // 项目已有原话记录的最大记录时间
            String maxTime = asyncTaskUtil.getMaxTimeFromMongo(projectId, field.getId());
            // 查询 clickhouse 的时间过滤条件
            ColumnDTO columnDTO = columnDTOMap.get(TIME);
            if (Objects.isNull(columnDTO)) {
                log.error("[Txtanls] 问卷项目 {} 无 time 字段", projectId);
                return;
            }
            List<Filter> timeFilters = clickhouseUtil.getTimeFilter(columnDTO, maxTime, beginTime);
            timeFilters.add(statusFilter);
            // 待查询的 clickhouse 列信息
            List<ClickhouseColumnDTO> searchColumn = clickhouseUtil.getSurveyClickhouseColumns(project, field, columnDTOMap);

            int skip = 0, size = LIMIT, progressCountNew = 0;
            List<TrainTextDTO> trainTextDTOList = new ArrayList<>();

            while (size == LIMIT) {
                // 查询clickhouse
                ClickHouseDTO clickHouseDTO = clickhouseUtil.getClickhouseData(worktableDTO.getId(), searchColumn, skip, timeFilters);
                if (Objects.isNull(clickHouseDTO) || CollectionUtils.isEmpty(clickHouseDTO.getAllRowsData())) {
                    log.warn("[Txtanls] {}-{} clickhouse返回结果为空", worktableDTO.getId(), field.getValue());
                    break;
                }
                if (!clickHouseDTO.getColNameList().contains(fieldName)) {
                    log.error("[Txtanls] {}-{} clickhouse返回结果无训练字段", worktableDTO.getId(), field.getValue());
                    break;
                }
                if (!clickHouseDTO.getColNameList().contains(timeFieldName)) {
                    log.error("[Txtanls] {}-{} clickhouse返回结果无时间字段", worktableDTO.getId(), field.getValue());
                    break;
                }
                size = clickHouseDTO.getAllRowsData().size();


                for (int index = 0; index < clickHouseDTO.getAllRowsData().size(); index++) {
                    Map<String, String> rowData = clickHouseDTO.getAllRowsData().get(index);
                    // 要训练的文本记录
                    String record = rowData.get(fieldName);
                    // 无用字符清理
                    record = TextUtil.text_clean(record);
                    if (!ObjectUtils.isEmpty(record)) {
                        // 训练时间字段
                        String time = asyncTaskUtil.formatTime(rowData.get(timeFieldName));
                        // 唯一标识时间字段
                        String collectionTime = rowData.get(COLLECTION_TIME);
                        trainTextDTOList.add(new TrainTextDTO(record, time, rowData.get(COLLECTION_ID), collectionTime));
                    }

                    // 到达设置的阈值, 批量训练
                    if (trainTextDTOList.size() == REQUEST_NUMBER) {
                        trainRecordData(project, field, trainTextDTOList, business, childrenViewPoint);
                        trainTextDTOList = new ArrayList<>();
                    }
                    // 到达设置的阈值, 更新训练进度
                    if ((progressCountNew += 1) == PROGRESS_COUNT) {
                        asyncTaskUtil.updateTrainProgress(projectId, field.getId(), PROGRESS_COUNT);
                        progressCountNew = 0;
                    }
                }

                skip += size;
            }

            // 兜底训练, 防止数据遗漏
            if (!CollectionUtils.isEmpty(trainTextDTOList)) {
                trainRecordData(project, field, trainTextDTOList, business, childrenViewPoint);
                // 更新训练进度
                asyncTaskUtil.updateTrainProgress(projectId, field.getId(), trainTextDTOList.size());
            }
        }
    }


    /**
     * 调用情感接口, 同时保存结果
     *
     * @param project           项目信息
     * @param textField         分析的字段
     * @param trainTextDTOList  训练的参数
     * @param business          项目对应行业信息
     * @param childrenViewPoint 项目观点信息
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/23 11:40
     */
    public void trainRecordData(ProjectDO project, ProjectFieldDO textField, List<TrainTextDTO> trainTextDTOList,
                                String business, List<ViewpointDO> childrenViewPoint) {
        String projectId = project.getId();

        // 将文本中的换行去除
        List<String> data = new ArrayList<>();
        for (TrainTextDTO trainTextDTO : trainTextDTOList) {
            String text = trainTextDTO.getText();
            data.add(text.replaceAll("\n", "").toLowerCase());
        }
        // 行业观点
//        IndustryCodeEnum business = asyncTaskUtil.getBusiness(industry);
        // 调用大数据数据分析接口获取文本及其中观点情感
        List<TextSentiment> textSentimentList = callAnalysisApiSpringRetry.callApi(data, business);

        if (CollectionUtils.isEmpty(textSentimentList) || textSentimentList.size() != trainTextDTOList.size()) {
            log.error("[Txtanls] 情感分析接口调用返回结果长度不匹配; ProjectId:{}; Result:{}", projectId, textSentimentList);
            return;
        }

        // 情感接口返回数据转换
        List<SentimentDTO> sentimentDTOList = asyncTaskUtil.responseParse(textSentimentList);
        if (CollectionUtils.isEmpty(sentimentDTOList) || sentimentDTOList.size() != trainTextDTOList.size()) {
            log.error("[Txtanls] 情感接口返回数据转换结果长度不匹配; ProjectId:{}; Result:{}", projectId, sentimentDTOList);
            return;
        }
        List<Map<String, Object>> sentiViews;
        int senti;
        for (int i = 0; i < trainTextDTOList.size(); i++) {
            TrainTextDTO trainTextDTO = trainTextDTOList.get(i);
            // 如果解析不了原话时间，就跳过
            if (Validator.isEmpty(AsyncTaskUtil.formatToNewTime(trainTextDTO.getTime()))) {
                log.warn("[Txtanls] trainRecordData 训练时间 {}", trainTextDTO.getTime());
                continue;
            }
            senti = sentimentDTOList.get(i).getTextSenti();
            sentiViews = sentimentDTOList.get(i).getOpinionSenti();

            // 返回的观点情感数据和已有观点数据进行匹配
            List<JoinInfo.Viewpoint> listViewPoint = asyncTaskUtil.viewPointConvert(childrenViewPoint, sentiViews);
            // 保存原话记录
            RecordDO recordDOEntity = asyncTaskUtil.saveRecord(project, trainTextDTO, senti, textField, null);
            // 保存训练结果
            asyncTaskUtil.saveJoinInfo(projectId, trainTextDTO, senti, recordDOEntity.getId(), listViewPoint);
        }
    }
}
